%%%-------------------------------------------------------------------
%%% File    : ptool_local.erl
%%% Author  :  <vjache>
%%% Description : 
%%%
%%% Created : 27 Apr 2009 by  <vjache>
%%%-------------------------------------------------------------------
-module(ptool_local).

-include("../include/ptool.hrl").
  
%% API
-export([
		 new/0,
		 new/1,
		 open/1,
		 obtain_acc/5,
		 create_acc/5,
		 send_acc/3,
		 finish_acc/2,
		 finish_all/1,
		 yield_acc/2,
		 yield_all/2,
		 yield_atwill_all/4,
		 close/1
]).
-export([test/0]).

%%====================================================================
%% API
%%====================================================================
new() ->
	new(fun(Item)->
				Item
		end).
new(KeyFun) ->
	#ptool{module=?MODULE,open_args=[KeyFun],key_fun=KeyFun}.

%%====================================================================
%% Callback API for module 'ptool'
%%====================================================================
open([KeyFun]) ->
	{_PrcDictName=make_ref(),KeyFun}.

obtain_acc({PrcDictName,KeyFun},Item,FoldFun,FinishFun,Acc) ->
	Key=KeyFun(Item),
	cache_ets:obtain(PrcDictName, Key, 
					 fun() ->
							 _Handle=create_acc({PrcDictName,KeyFun},node(),
											   fun(Itm,AccIn)->
													   FoldFun(Key,Itm,AccIn)
											   end,
											   fun(AccIn)->
													   FinishFun(Key,AccIn)
											   end,Acc)
					 end).

create_acc(_PToolState={PrcDictName,_KeyFun},Node,FoldFun,FinishFun,Acc) ->
	CallerPid=self(),
	{'$acc_prc',spawn_link(Node,fun()->
										acc_loop(PrcDictName,FoldFun,FinishFun,Acc,CallerPid)
						   end)}.
send_acc(_PToolState,{'$acc_prc',Pid},Item) ->
	Pid!{'$acc_prc_item',Item},
	ok.
finish_acc(_PToolState,{'$acc_prc',Pid}) ->
	Pid!{'$acc_prc_finish',self()},
	_Promise={'$acc_prc_promise',Pid}.
finish_all({PrcDictName,_KeyFun}=State) ->
	R=cache_ets:foldl(PrcDictName,
					fun({_K,Handle},Proms)->
							[finish_acc(State,Handle)|Proms]
					end, _Proms=[]),
	ok=cache_ets:cleanup(PrcDictName),
	R.
yield_acc(_PToolState={PrcDictName,_KeyFun},{'$acc_prc_promise',Pid}=_Promise) ->
	receive
		{'$acc_prc_result',PrcDictName,Pid,Acc} ->
			Acc
	end.

yield_all(PToolState,Promises) ->
	yield_all(PToolState,Promises,[]).

yield_all(_PToolState,[],Accs) ->
	Accs;
yield_all(PToolState={PrcDictName,_KeyFun},Promises,Accs) ->
	receive
		{'$acc_prc_result',PrcDictName,Pid,Acc} ->
			Promises2=lists:keydelete(
						Pid, 2, Promises),
			yield_all(PToolState,Promises2,[Acc|Accs])
	end.

yield_atwill_all(_PToolState,[],_ConsumeFun,ConsumerState) ->
	ConsumerState;
yield_atwill_all(PToolState={PrcDictName,_KeyFun},Promises,ConsumeFun,ConsumerState) ->
	receive
		{'$acc_prc_result',PrcDictName,Pid,Acc} ->
			Promises2=lists:keydelete(
						Pid, 2, Promises),
			yield_atwill_all(PToolState,Promises2,ConsumeFun,ConsumeFun(Acc,ConsumerState))
	end.

close({PrcDictName,_KeyFun}=_State) ->
	ok=cache_ets:cleanup(PrcDictName).

%%====================================================================
%% Internal functions
%%====================================================================

acc_loop(PrcDictName,FoldFun,FinishFun,Acc,CallerPid) ->
	receive
		{'$acc_prc_item',Item} ->
			acc_loop(PrcDictName,FoldFun,FinishFun,FoldFun(Item,Acc),CallerPid);
		{'$acc_prc_finish',CallerPid} ->
			CallerPid!{'$acc_prc_result',PrcDictName,self(),FinishFun(Acc)}
	end.

%%====================================================================
%% Test functions
%%====================================================================

test() ->
	List=[
		  {predicate,"pred","subj","obj","graph"},
		  {predicate,"pred","subj1","obj","graph"},
		  {predicate,"pred","subj2","obj--","graph"},
		  {predicate,"pred","subj3","obj74","graph"},
		  {predicate,"pred","subj4","obj45","graph"},
		  {predicate,"pred","subj5","obj%%","graph"},
		  {predicate,"pred","subj6","obj","graph"},
		  {predicate,"pred1","subj","obj","graph"},
		  {predicate,"pred1","subj1","obj","graph"},
		  {predicate,"pred1","subj2","obj--","graph"},
		  {predicate,"pred2","subj3","obj74","graph"},
		  {predicate,"pred2","subj4","obj45","graph"},
		  {predicate,"pred2","subj5","obj%%","graph"},
		  {predicate,"pred2","subj6","obj","graph"},
		  {predicate,"pred","subj","obj","graph"},
		  {predicate,"pred","subj1","obj","graph"},
		  {predicate,"pred","subj2","obj--","graph"},
		  {predicate,"pred","subj3","obj74","graph"},
		  {predicate,"pred","subj4","obj45","graph"},
		  {predicate,"pred","subj5","obj%%","graph"},
		  {predicate,"pred","subj6","obj","graph"},
		  {predicate,"pred1","subj","obj","graph"},
		  {predicate,"pred1","subj1","obj","graph"},
		  {predicate,"pred1","subj2","obj--","graph"},
		  {predicate,"pred2","subj3","obj74","graph"},
		  {predicate,"pred2","subj4","obj45","graph"},
		  {predicate,"pred2","subj5","obj%%","graph"},
		  {predicate,"pred2","subj6","obj","graph"},
		  {predicate,"pred","subj","obj","graph"},
		  {predicate,"pred","subj1","obj","graph"},
		  {predicate,"pred","subj2","obj--","graph"},
		  {predicate,"pred","subj3","obj74","graph"},
		  {predicate,"pred","subj4","obj45","graph"},
		  {predicate,"pred","subj5","obj%%","graph"},
		  {predicate,"pred","subj6","obj","graph"},
		  {predicate,"pred1","subj","obj","graph"},
		  {predicate,"pred1","subj1","obj","graph"},
		  {predicate,"pred1","subj2","obj--","graph"},
		  {predicate,"pred2","subj3","obj74","graph"},
		  {predicate,"pred2","subj4","obj45","graph"},
		  {predicate,"pred2","subj5","obj%%","graph"},
		  {predicate,"pred2","subj6","obj","graph"},
		  {predicate,"pred","subj","obj","graph"},
		  {predicate,"pred","subj1","obj","graph"},
		  {predicate,"pred","subj2","obj--","graph"},
		  {predicate,"pred","subj3","obj74","graph"},
		  {predicate,"pred","subj4","obj45","graph"},
		  {predicate,"pred","subj5","obj%%","graph"},
		  {predicate,"pred","subj6","obj","graph"},
		  {predicate,"pred1","subj","obj","graph"},
		  {predicate,"pred1","subj1","obj","graph"},
		  {predicate,"pred1","subj2","obj--","graph"},
		  {predicate,"pred2","subj3","obj74","graph"},
		  {predicate,"pred2","subj4","obj45","graph"},
		  {predicate,"pred2","subj5","obj%%","graph"},
		  {predicate,"pred2","subj6","obj","graph"},
		  {predicate,"pred3","subj","obj","graph"},
		  {predicate,"pred3","subj1","obj","graph"},
		  {predicate,"pred3","subj2","obj--","graph"},
		  {predicate,"pred3","subj3","obj74","graph"},
		  {predicate,"pred3","subj4","obj45","graph"},
		  {predicate,"pred3","subj5","obj%%","graph"},
		  {predicate,"pred3","subj6","obj","graph"},
		  {predicate,"pred31","subj","obj","graph"},
		  {predicate,"pred31","subj1","obj","graph"},
		  {predicate,"pred31","subj2","obj--","graph"},
		  {predicate,"pred32","subj3","obj74","graph"},
		  {predicate,"pred32","subj4","obj45","graph"},
		  {predicate,"pred32","subj5","obj%%","graph"},
		  {predicate,"pred23","subj6","obj","graph"}
		 ],
	PTool=ptool:open(new(fun(T)->
				 element(2,T)
		 end)),
	ptool:pkeyfold(PTool,
		 fun(_Key,Item,Acc)->
				 [Item|Acc]
		 end,
		 fun(Key,Acc)->
				 {Key,lists:foldl(fun(_I,Count)->
										  Count+1
								  end, 0, Acc)}
		 end,
		 [],List).


